package com.github.binarylei.hello_world.topology;

import com.github.binarylei.hello_world.bolt.PrintBolt;
import com.github.binarylei.hello_world.bolt.WriteBolt;
import com.github.binarylei.hello_world.spout.PWSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;

/**
 * 工作进程，并行度，任务数设置
 * @author: leigang
 * @version: 2018-04-07
 */
public class PWTopology2 {

    /**
     * 1. 首先设置了2个工作进程(也就是2个 JVM)
     * 2. 然后我们设置了 spout 的并行度为2(产生2个执行器和2个任务)
     * 3. 第一个 bolt 的并行度为 2 并且指定任务数为 4(产生 2 个执行器和 4 个任务)
     * 4. 第二个 bolt 的并行度为 6(产生 6 个执行器和 6 个任务)
     * 5. 因此:该拓扑程序共有 2 个工作进程(worker)，2+2+6=10 个执行器(executor)，2+4+6=12 个任务(task)。
     *    每个工作进程可以领取到12/2=6个任务。默认情况下一个执行器执行一个任务，但如果指定了任务的数目。则任务会平均分配到执行器中。
     */
    public static void main(String[] args) throws InterruptedException {
        Config cfg = new Config();
        //1. 设置使用两个工作进程
        cfg.setNumWorkers(2);
        cfg.setDebug(true);
        TopologyBuilder builder = new TopologyBuilder();
        //2. 设置spout的并行度和任务数(产生2个执行器和2个任务)
        builder.setSpout("spout", new PWSpout(), 2);
        //3. 设置blot的并行度和任务数(产生2个执行器和4个任务)
        builder.setBolt("print-bolt", new PrintBolt(), 2)
                .shuffleGrouping("spout").setNumTasks(4);
        //4. 设置blot的并行度和任务数(产生6个执行器和6个任务)
        builder.setBolt("write-bolt", new WriteBolt(), 6)
                .shuffleGrouping("print-bolt");

        //1. 本地模式
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("top1", cfg, builder.createTopology());
        Thread.sleep(20 * 1000);
        cluster.killTopology("top1");
        cluster.shutdown();

        //2. 集群模式
        //cfg.setNumWorkers(1);
        //cfg.setDebug(true);
        //StormSubmitter.submitTopology("top1", cfg, builder.createTopology());

    }
}
